package com.atguigu.edu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogApp {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String topicName = "topic_log";
        String groupID = "base_log_app";
        DataStreamSource<String> topicLogStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupID));
        OutputTag<String> outputTag = new OutputTag<String>("Dirty") {};

        SingleOutputStreamOperator<JSONObject> process = topicLogStream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try{
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                }catch (Exception e){
                    ctx.output(outputTag,value);
                }

            }
        });
        DataStream<String> sideOutput = process.getSideOutput(outputTag);

        // TODO 5 修正is_new字段
        KeyedStream<JSONObject, String> jsonObjectStringKeyedStream = process.keyBy(json -> json.getJSONObject("common").getString("mid"));

        SingleOutputStreamOperator<JSONObject> process1 = jsonObjectStringKeyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

            private ValueState<String> firstVisitDt;

            @Override
            public void open(Configuration parameters) throws Exception {
                firstVisitDt = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstVisitDt", String.class));
            }

            @Override
            public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                String string = value.getJSONObject("common").getString("is_new");
                String value1 = firstVisitDt.value();
                Long ts = value.getLong("ts");
                String s = DateFormatUtil.toDate(ts);

                if ("1".equals(string)) {
                    if (firstVisitDt == null) {
                        firstVisitDt.update(s);
                    } else {
                        if (!firstVisitDt.equals(s)) {
                            string = "0";
                            value.getJSONObject("common").put("is_new", string);
                        }
                    }
                } else {
                    if (firstVisitDt == null) {
                        String s1 = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                        firstVisitDt.update(s1);
                    }
                }
                out.collect(value);
            }
        });

        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {
        };

        SingleOutputStreamOperator<String> process2 = process1.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                JSONObject err = value.getJSONObject("err");
                if (err != null) {
                    ctx.output(errorTag, value.toJSONString());
                }

                value.remove("err");

                JSONObject start = value.getJSONObject("start");
                if (start != null) {
                    ctx.output(startTag, value.toJSONString());
                } else {
                    JSONObject page = value.getJSONObject("page");
                    JSONObject common = value.getJSONObject("common");
                    Long ts = value.getLong("ts");
                    JSONArray displays = value.getJSONArray("displays");


                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            JSONObject displayObj = new JSONObject();
                            displayObj.put("display", display);
                            displayObj.put("common", common);
                            displayObj.put("page", page);
                            displayObj.put("ts", ts);
                            ctx.output(displayTag, displayObj.toJSONString());
                        }
                    }

                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject jsonObject = actions.getJSONObject(i);
                            JSONObject jsonObject1 = new JSONObject();
                            jsonObject1.put("action", jsonObject);
                            jsonObject1.put("common", common);
                            jsonObject1.put("page", page);
                            ctx.output(actionTag, jsonObject1.toJSONString());
                        }
                    }

                    value.remove("displays");
                    value.remove("actions");
                    out.collect(value.toString());
                }

            }
        });

        DataStream<String> startDS = process2.getSideOutput(startTag);
        DataStream<String> displayDS = process2.getSideOutput(displayTag);
        DataStream<String> actionDS = process2.getSideOutput(actionTag);
        DataStream<String> errorDS = process2.getSideOutput(errorTag);

        // 7.2 定义不同日志输出到 Kafka 的主题名称
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";


        process2.addSink(KafkaUtil.getKafkaProducer(page_topic));
        startDS.addSink(KafkaUtil.getKafkaProducer(start_topic));
        displayDS.addSink(KafkaUtil.getKafkaProducer(display_topic));
        actionDS.addSink(KafkaUtil.getKafkaProducer(action_topic));
        errorDS.addSink(KafkaUtil.getKafkaProducer(error_topic));

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("env.execute()执行失败");
        }
    }
}
